package cn.middle.remote.edge;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.middle.remote.Buffer;
import cn.middle.remote.enums.CallState;
import cn.middle.remote.message.BaseSignal;
import cn.middle.remote.message.CallSignal;
import cn.middle.remote.message.HeartbeatPongSignal;
import cn.middle.remote.message.LoginResultSignal;
import cn.middle.remote.message.ReplyMessageSignal;
import cn.middle.remote.message.SignalHelper;
import cn.middle.remote.util.NettyAttributeUtil;
import io.netty.channel.ChannelHandlerContext;

public class NettyEdgeProcessHandler extends EdgeProcessingHandler {
	
	private static final Logger log = LoggerFactory.getLogger(NettyEdgeProcessHandler.class);
	
	ExecutorService  executors = Executors.newFixedThreadPool(10);

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		listener.connectLost(NettyAttributeUtil.clientId(ctx.channel()),1);
		log.error("user-close -> {}" ,ctx.channel().remoteAddress());
	}

	@Override
	public void received(ChannelHandlerContext ctx, BaseSignal msg) throws Exception {
		executors.execute(() ->{
			switch(msg.getMessageType()) {
			case SignalHelper.CALL:
				reply(ctx, (CallSignal)msg);
				break;
			case SignalHelper.PONG:
				pong((HeartbeatPongSignal)msg);
				break;
			case SignalHelper.LOGIN_RESULT:
				login(ctx, (LoginResultSignal)msg);
				break;
			case SignalHelper.ACK:
				ack(msg.getId());
				break;
			}
		});
	}
	
	private void reply(ChannelHandlerContext ctx,CallSignal msg) {
		listener.hairDown(obj ->{
			byte[] data =  serialization.encode(obj);
			ReplyMessageSignal signal = new ReplyMessageSignal();
			signal.setId(msg.getId());
			signal.setRemoteAddressClientId(msg.getRemoteAddressClientId());
			signal.setData(data);
			signal.setCode(CallState.SUCCESS.getCode());
			signal.setInterfaceName(msg.getInterfaceName());
			ctx.channel().writeAndFlush(signal);
		}, msg.getInterfaceName(), new Buffer(msg.getId(),msg.getData(),serialization));
	}
	
	private void login(ChannelHandlerContext ctx,LoginResultSignal signal) {
		log.info("login : " + signal.getId() + ": " + signal.getCode() + ": " + signal.getMessage());
		listener.loginResult(signal.getCode(), signal.getMessage());
	}
	
	private void ack(String msgId) {
		log.info("ack -> {}" ,msgId );
		listener.ack(msgId);
	}
	
	private void pong(HeartbeatPongSignal msg) {
		log.info(msg.getId() + " : " + HeartbeatPongSignal.PONG);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		log.error(cause.getMessage(),cause);
		listener.exception(cause);
	}

}
